package com.example.multiaccountfetcher.queue;

import com.example.multiaccountfetcher.config.AccountConfig;
import com.example.multiaccountfetcher.model.AccountRequest;
import com.example.multiaccountfetcher.service.DataFetchService;
import com.example.multiaccountfetcher.strategy.DataFetchStrategy;
import com.example.multiaccountfetcher.strategy.DataFetchStrategyFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Slf4j
public class RequestQueueManager {
    
    @Autowired
    private AccountConfig accountConfig;
    
    @Autowired
    private DataFetchService dataFetchService;
    
    @Autowired
    private DataFetchStrategyFactory strategyFactory;
    
    // 存储每个账号的上次请求时间
    private final Map<String, LocalDateTime> lastRequestTimes = new ConcurrentHashMap<>();
    
    // 存储每个账号和接口组合的拉取次数
    private final Map<String, Integer> fetchCounts = new ConcurrentHashMap<>();
    
    // 待处理的请求队列
    private final BlockingQueue<AccountRequest> requestQueue;
    
    // 正在处理的请求计数器
    private final AtomicInteger activeRequests = new AtomicInteger(0);
    
    // 调度线程池
    private ExecutorService schedulerExecutor;
    
    // 工作线程池
    private ExecutorService workerExecutor;
    
    public RequestQueueManager() {
        // 初始化请求队列，容量从配置中获取
        this.requestQueue = new LinkedBlockingQueue<>(10000); // 默认容量，会在初始化时更新
    }
    
    @PostConstruct
    public void init() {
        // 更新队列容量
        int capacity = accountConfig.getQueueBufferSize();
        if (capacity > 0 && capacity != requestQueue.size()) {
            // 创建新的队列并转移元素
            BlockingQueue<AccountRequest> newQueue = new LinkedBlockingQueue<>(capacity);
            requestQueue.drainTo(newQueue);
            requestQueue.addAll(newQueue);
        }
        
        // 初始化调度线程池
        schedulerExecutor = Executors.newSingleThreadExecutor(r -> {
            Thread thread = new Thread(r, "request-scheduler");
            thread.setDaemon(true);
            return thread;
        });
        
        // 初始化工作线程池
        int maxThreads = accountConfig.getMaxConcurrentAccounts();
        workerExecutor = new ThreadPoolExecutor(
                Math.min(10, maxThreads), // 核心线程数
                maxThreads, // 最大线程数
                60L, TimeUnit.SECONDS, // 空闲线程存活时间
                new LinkedBlockingQueue<>(capacity), // 工作队列
                r -> {
                    Thread thread = new Thread(r, "request-worker-");
                    thread.setDaemon(true);
                    return thread;
                },
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
        );
        
        // 启动调度线程
        schedulerExecutor.submit(this::processQueue);
        
        log.info("Request queue manager initialized with max concurrent accounts: {}, queue capacity: {}", 
                 maxThreads, capacity);
    }
    
    /**
     * 添加请求到队列
     */
    public boolean addRequest(AccountRequest request) {
        if (request == null || request.getAccountId() == null) {
            log.error("Invalid request: null or missing accountId");
            return false;
        }
        
        // 设置请求时间
        request.setRequestTime(LocalDateTime.now());
        
        boolean added = requestQueue.offer(request);
        if (added) {
            log.debug("Request added to queue for account: {}", request.getAccountId());
        } else {
            log.warn("Queue is full, request rejected for account: {}", request.getAccountId());
        }
        
        return added;
    }
    
    /**
     * 处理队列中的请求
     */
    private void processQueue() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                // 检查是否可以处理更多请求
                if (activeRequests.get() >= accountConfig.getMaxConcurrentAccounts()) {
                    Thread.sleep(10); // 短暂休眠避免CPU空转
                    continue;
                }
                
                // 尝试获取下一个可处理的请求
                AccountRequest request = findNextProcessableRequest();
                if (request != null) {
                    processRequest(request);
                } else {
                    // 没有可处理的请求，等待一小段时间
                    Thread.sleep(10);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.info("Request scheduler interrupted");
                break;
            } catch (Exception e) {
                log.error("Error processing request queue", e);
            }
        }
    }
    
    /**
     * 查找下一个可处理的请求
     * 实现智能排队算法，优先选择等待时间最长且满足间隔要求的请求
     */
    private AccountRequest findNextProcessableRequest() {
        if (requestQueue.isEmpty()) {
            return null;
        }
        
        LocalDateTime now = LocalDateTime.now();
        
        // 收集所有满足间隔要求的请求
        List<AccountRequest> eligibleRequests = new ArrayList<>();
        Iterator<AccountRequest> iterator = requestQueue.iterator();
        
        while (iterator.hasNext()) {
            AccountRequest request = iterator.next();
            String accountId = request.getAccountId();
            String interfaceType = request.getInterfaceType();
            
            // 获取接口类型对应的策略和间隔时间
            int minIntervalSeconds = accountConfig.getMinIntervalSeconds();
            if (interfaceType != null) {
                minIntervalSeconds = strategyFactory.getStrategy(interfaceType)
                    .map(DataFetchStrategy::getRequestIntervalSeconds)
                    .orElse(minIntervalSeconds);
            }
            
            // 构建账号和接口的唯一键
            String key = accountId + ":" + (interfaceType != null ? interfaceType : "default");
            LocalDateTime lastRequestTime = lastRequestTimes.get(key);
            
            // 检查是否满足间隔要求
            if (lastRequestTime == null || 
                ChronoUnit.SECONDS.between(lastRequestTime, now) >= minIntervalSeconds) {
                eligibleRequests.add(request);
            }
        }
        
        if (eligibleRequests.isEmpty()) {
            return null;
        }
        
        // 按请求时间排序，优先处理最早的请求
        eligibleRequests.sort(Comparator.comparing(AccountRequest::getRequestTime));
        
        // 从队列中移除选中的请求
        AccountRequest selectedRequest = eligibleRequests.get(0);
        requestQueue.remove(selectedRequest);
        
        return selectedRequest;
    }
    
    /**
     * 处理单个请求
     */
    private void processRequest(AccountRequest request) {
        String accountId = request.getAccountId();
        String interfaceType = request.getInterfaceType();
        
        // 构建账号和接口的唯一键
        String key = accountId + ":" + (interfaceType != null ? interfaceType : "default");
        
        // 更新活跃请求计数
        activeRequests.incrementAndGet();
        
        try {
            // 更新上次请求时间
            lastRequestTimes.put(key, LocalDateTime.now());
            
            // 获取当前账号和接口的拉取次数
            int currentCount = fetchCounts.getOrDefault(key, 0);
            fetchCounts.put(key, currentCount + 1);
            
            log.debug("Processing request for account: {}, interfaceType: {}", accountId, interfaceType);
            
            // 确定使用的策略
            if (interfaceType != null) {
                Optional<DataFetchStrategy> strategyOptional = strategyFactory.getStrategy(interfaceType);
                if (strategyOptional.isPresent()) {
                    DataFetchStrategy strategy = strategyOptional.get();
                    processWithStrategy(request, strategy, key, currentCount);
                } else {
                    // 策略不存在，使用默认的数据拉取服务
                    processWithDefaultService(request);
                }
            } else {
                // 没有指定接口类型，使用默认的数据拉取服务
                processWithDefaultService(request);
            }
        } catch (Exception e) {
            log.error("Failed to submit request for account: {}", accountId, e);
            activeRequests.decrementAndGet();
            if (request.getCallback() != null) {
                request.getCallback().onFailure(accountId, e.getMessage());
            }
        }
    }
    
    /**
     * 使用指定的策略处理请求
     */
    private void processWithStrategy(AccountRequest request, DataFetchStrategy strategy, String key, int currentCount) {
        String accountId = request.getAccountId();
        
        try {
            // 使用策略拉取数据
            CompletableFuture.supplyAsync(() -> {
                try {
                    return strategy.fetchData(accountId, request.getParams())
                            .get(accountConfig.getFetchTimeoutSeconds(), TimeUnit.SECONDS);
                } catch (Exception e) {
                    throw new RuntimeException("Failed to fetch data with strategy", e);
                }
            }, workerExecutor)
            .thenAccept(result -> {
                try {
                    if (request.getCallback() != null) {
                        request.getCallback().onSuccess(accountId, result);
                    }
                    
                    // 检查是否需要继续拉取数据
                    if (strategy.shouldContinueFetching(accountId, currentCount + 1)) {
                        // 创建新的请求继续拉取
                        AccountRequest newRequest = new AccountRequest();
                        newRequest.setAccountId(accountId);
                        newRequest.setRequestTime(LocalDateTime.now());
                        newRequest.setParams(request.getParams());
                        newRequest.setCallback(request.getCallback());
                        newRequest.setInterfaceType(request.getInterfaceType());
                        requestQueue.offer(newRequest);
                    }
                } finally {
                    activeRequests.decrementAndGet();
                }
            })
            .exceptionally(ex -> {
                try {
                    log.error("Error processing request with strategy for account: {}", accountId, ex);
                    if (request.getCallback() != null) {
                        request.getCallback().onFailure(accountId, ex.getMessage());
                    }
                } finally {
                    activeRequests.decrementAndGet();
                }
                return null;
            });
        } catch (Exception e) {
            log.error("Failed to process with strategy for account: {}", accountId, e);
            activeRequests.decrementAndGet();
            if (request.getCallback() != null) {
                request.getCallback().onFailure(accountId, e.getMessage());
            }
        }
    }
    
    /**
     * 使用默认的数据拉取服务处理请求
     */
    private void processWithDefaultService(AccountRequest request) {
        String accountId = request.getAccountId();
        
        try {
            // 提交到工作线程池处理
            CompletableFuture.supplyAsync(() -> {
                try {
                    return dataFetchService.fetchData(accountId, request.getParams())
                            .get(accountConfig.getFetchTimeoutSeconds(), TimeUnit.SECONDS);
                } catch (Exception e) {
                    throw new RuntimeException("Failed to fetch data", e);
                }
            }, workerExecutor)
            .thenAccept(result -> {
                try {
                    if (request.getCallback() != null) {
                        request.getCallback().onSuccess(accountId, result);
                    }
                } finally {
                    activeRequests.decrementAndGet();
                }
            })
            .exceptionally(ex -> {
                try {
                    log.error("Error processing request for account: {}", accountId, ex);
                    if (request.getCallback() != null) {
                        request.getCallback().onFailure(accountId, ex.getMessage());
                    }
                } finally {
                    activeRequests.decrementAndGet();
                }
                return null;
            });
        } catch (Exception e) {
            log.error("Failed to submit request for account: {}", accountId, e);
            activeRequests.decrementAndGet();
            if (request.getCallback() != null) {
                request.getCallback().onFailure(accountId, e.getMessage());
            }
        }
    }
    
    /**
     * 获取队列状态
     */
    public QueueStatus getQueueStatus() {
        return new QueueStatus(
                requestQueue.size(),
                activeRequests.get(),
                lastRequestTimes.size()
        );
    }
    
    /**
     * 队列状态类
     */
    public static class QueueStatus {
        private final int queueSize;
        private final int activeRequests;
        private final int trackedAccounts;
        
        public QueueStatus(int queueSize, int activeRequests, int trackedAccounts) {
            this.queueSize = queueSize;
            this.activeRequests = activeRequests;
            this.trackedAccounts = trackedAccounts;
        }
        
        public int getQueueSize() { return queueSize; }
        public int getActiveRequests() { return activeRequests; }
        public int getTrackedAccounts() { return trackedAccounts; }
        
        @Override
        public String toString() {
            return "QueueStatus{" +
                    "queueSize=" + queueSize +
                    ", activeRequests=" + activeRequests +
                    ", trackedAccounts=" + trackedAccounts +
                    '}';
        }
    }
    
    @PreDestroy
    public void shutdown() {
        log.info("Shutting down request queue manager");
        
        // 关闭调度线程池
        if (schedulerExecutor != null) {
            schedulerExecutor.shutdownNow();
        }
        
        // 关闭工作线程池
        if (workerExecutor != null) {
            workerExecutor.shutdownNow();
        }
        
        log.info("Request queue manager shutdown complete");
    }
    
}